package schedule;

import batchMessage.NetworkUtils;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.MixAll;
import com.alibaba.rocketmq.common.UtilAll;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageAccessor;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;

public class ScheduleMessageConsumer {
    static final Logger logger = LoggerFactory.getLogger(ScheduleMessageConsumer.class);


    // 自定义模式
    private static DateTimeFormatter formater = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    public static final String TEST_TOPIC_2 = "TestTop3";
    public static final int MAX_RECONSUME_TIMES = 6;
    public static final String NAMESRV_ADDR = "127.0.0.1:9876";//test
//    public static final String NAMESRV_ADDR = "10.252.193.20:9876;10.252.193.18:9876";//test
    //    public static final String NAMESRV_ADDR = "10.0.75.153:9876;10.0.75.154:9876";//pre
//    public static final String NAMESRV_ADDR = "10.0.79.103:9876;10.0.79.104:9876";
    public static AtomicInteger index = new AtomicInteger();


    private static void consumeAndFail() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(TEST_TOPIC_2 + "-schedual-message");
        try {
            consumer.subscribe(TEST_TOPIC_2, "*");
            consumer.setNamesrvAddr(NAMESRV_ADDR);
            consumer.setMaxReconsumeTimes(4);

            //fixme 会导致消费不成功，或者重复的情况；
            consumer.setClientIP(NetworkUtils.getIpAddress0());
            consumer.setInstanceName(UUID.randomUUID().toString());
            consumer.registerMessageListener(new MessageListenerConcurrently() {

                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages,
                                                                ConsumeConcurrentlyContext context) {
                    for (MessageExt message : messages) {
                        System.out.println(LocalDateTime.now().format(formater) + " , Receive message (msgId=" + message.getMsgId() + " , time: "
                                + (System.currentTimeMillis() - message.getStoreTimestamp()));
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }


    private static void consume() throws MQClientException {
        logger.info("test");
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(TEST_TOPIC_2 + "-schedual-message");
        consumer.subscribe(TEST_TOPIC_2, "*");
        consumer.setNamesrvAddr(NAMESRV_ADDR);
        consumer.setPullBatchSize(100);
        consumer.setConsumeThreadMax(30);
        consumer.setConsumeThreadMin(5);
        consumer.setMessageModel(MessageModel.CLUSTERING);
        consumer.setMaxReconsumeTimes(MAX_RECONSUME_TIMES);

        //fixme 会导致消费不成功，或者重复的情况；
        consumer.setClientIP(NetworkUtils.getIpAddress0());
        consumer.setInstanceName(UUID.randomUUID().toString());

        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
                try {
                    for (MessageExt message : messages) {
                        System.out.println(index.incrementAndGet() + ", time=" + System.currentTimeMillis() + " =Receive message= " + message.getTopic() + " ,delay:"
                                + message.getDelayTimeLevel() + " , times:" + message.getReconsumeTimes() + " , msgId=" + message.getMsgId());
                    }
                    if (true) {
                        // throw new RuntimeException("");
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    //sendBack(consumer, messages.get(0));
                }
                //FIXME 本身会自动默认从3级别重试；
                System.out.println(index.get() + "=retry");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
    }

    private static void sendBack(DefaultMQPushConsumer consumer, MessageExt message) {
        System.out.println("=sendBack=>" + message.getMsgId());
        try {
            MessageExt msg = new MessageExt();
            msg.setBody("test".getBytes());
            msg.setMsgId(message.getMsgId());
            msg.setTopic(TEST_TOPIC_2);
            msg.setReconsumeTimes(message.getReconsumeTimes() + message.getDelayTimeLevel());
//12:39
//            consumer.sendMessageBack(msg, message.getDelayTimeLevel() + 1);

            Message newMsg = new Message(MixAll.getRetryTopic(consumer.getConsumerGroup()), msg.getBody());
            MessageAccessor.setProperties(newMsg, msg.getProperties());
            String originMsgId = MessageAccessor.getOriginMessageId(msg);
            MessageAccessor.setOriginMessageId(newMsg, UtilAll.isBlank(originMsgId) ? msg.getMsgId() : originMsgId);
            newMsg.setFlag(msg.getFlag());
            MessageAccessor.putProperty(newMsg, MessageConst.PROPERTY_RETRY_TOPIC, msg.getTopic());
            MessageAccessor.setReconsumeTime(newMsg, String.valueOf(message.getReconsumeTimes() + 1));
            MessageAccessor.setMaxReconsumeTimes(newMsg, MAX_RECONSUME_TIMES + "");
            newMsg.setDelayTimeLevel(message.getDelayTimeLevel() + 1);
            consumer.getDefaultMQPushConsumerImpl().getmQClientFactory().getDefaultMQProducer().send(newMsg);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    public static void main(String[] args) throws MQClientException {
        consumeAndFail();

    }
}
